package org.apache.zookeeper.server.paxos2;

import java.io.File;
import java.io.FilenameFilter;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.Map.Entry;


public class MetaStorage {

    private final String path;
    private final String prefix;
    private int count = 0;

    private TreeMap<Long, byte[]> data = new TreeMap<Long, byte[]>();
    private MergeThread mergeThread = new MergeThread();

    private final LinkedList<MetaFile> fileList = new LinkedList<MetaFile>();
    private final Object lock = new Object();

    public MetaStorage(String path, String prefix) {
        this.path = path;
        this.prefix = prefix;
        readFiles();
        mergeThread.start();
    }

    private void readFiles() {
        File[] files = new File(path).listFiles(new MetaFileFilter());
        TreeMap<Integer, MetaFile> sortFiles = new TreeMap<Integer, MetaFile>();
        for (File file : files) {
            String fileName = file.getAbsolutePath();
            if (MetaFile.checkMetaFileConsistency(fileName)) {
                MetaFile metaFile = new MetaFile(fileName, true);
                String tmp = file.getName();
                int index = tmp.indexOf('.');
                int logIndex = Integer.parseInt(tmp.substring(index + 1));
                sortFiles.put(logIndex, metaFile);
            }
        }
        for (MetaFile m : sortFiles.values()) {
            fileList.addLast(m);
        }
        if (sortFiles.size() > 0)
            count = sortFiles.lastKey() + 1;
    }


    private class MetaFileFilter implements FilenameFilter {

        public boolean accept(File dir, String name) {
            if (name.startsWith(prefix) && !name.endsWith("index"))
                return true;
            else
                return false;
        }
    }


    public void flushFile() {
        synchronized (fileList) {
            synchronized (lock) {
                if (data.size() == 0)
                    return;
                String fileName = path + File.separator + prefix
                        + "." + count;
                System.out.println("write to " + fileName);
                MetaFile file = new MetaFile(fileName, false);
                for (Entry<Long, byte[]> e : data.entrySet()) {
                    file.write(e.getKey(), e.getValue());
                }
                file.close();
                count++;

                fileList.addLast(new MetaFile(fileName, true));
                fileList.notifyAll();
                while (fileList.size() > 40) {
                    try {
                        fileList.wait();
                    }
                    catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                data.clear();
            }
        }
    }

    public void close() {
        flushFile();
        mergeThread.terminate();
    }


    public void write(long key, byte[] value) {
        if (value.length != BlockMeta.BLOCK_META_SIZE) {
            throw new RuntimeException("meta size does not match");
        }
        //System.out.println("put " + key + " " + Metadata.bytesToLong(value));
        boolean needFlush = false;
        synchronized (lock) {
            data.put(key, value);
            if (data.size() == Configuration.metaBufferSize)
                needFlush = true;
        }
        if (needFlush)
            flushFile();
    }

    public byte[] read(long key) {
        //throw new RuntimeException("not implemented");
        synchronized (fileList) {
            synchronized (lock) {
                //System.out.println("read "+key+" from "+this);
                if (data.containsKey(key))
                    return data.get(key);
            }
            Iterator<MetaFile> iter = fileList.descendingIterator();
            while (iter.hasNext()) {
                MetaFile f = iter.next();
                byte[] value = f.read(key);
                if (value != null)
                    return value;
            }
        }
        return null;
    }

    public interface MetaFinder {
        public boolean isOK(Entry<Long, byte[]> entry);
    }

    @SuppressWarnings("unchecked")
    public void find(MetaFinder finder, Storage storage, int sid) {
        LinkedList<Entry<Long, byte[]>> ret = new LinkedList<Entry<Long, byte[]>>();
        this.flushFile();
        synchronized (fileList) {
            Iterator<Entry<Long, byte[]>>[] iters = new Iterator[fileList.size()];
            int index = 0;
            for (MetaFile file : fileList) {
                iters[index++] = file.iterator();
            }
            MergeIterator mergeIter = new MergeIterator(iters);
            while (mergeIter.hasNext()) {
                Entry<Long, byte[]> e = mergeIter.next();
                if (finder == null || finder.isOK(e)) {
                    ret.addLast(e);
                    if (ret.size() >= 1000000) {
                        storage.sendStartSyncReply(ret, false, sid);
                        ret.clear();
                    }
                }
            }
            storage.sendStartSyncReply(ret, true, sid);
            ret.clear();

        }
    }

    public void print() {
    }


    @SuppressWarnings("unchecked")
    public void mergeFile() {
	if(Configuration.isInRecovery()||Configuration.isBusy()){
	    System.out.println("System busy: give up this merge");
	    return;
	}
        Configuration.setBusy(true);
        LinkedList<MetaFile> files = null;
        synchronized (fileList) {
            if (fileList.size() < 2) {
                System.out.println("no need to merge");
                return;
            }
            files = (LinkedList<MetaFile>) fileList.clone();
        }
        String lastFile = files.getLast().getPath();
        System.out.println("merge " + files + " to " + lastFile);
        Iterator<Entry<Long, byte[]>>[] iters = new Iterator[fileList.size()];
        int index = 0;
        MetaFile[] metaFiles = new MetaFile[files.size()];
        for (MetaFile file : files) {
            metaFiles[index] = file;
            iters[index] = metaFiles[index].iterator();
            index++;
        }
        MergeIterator mergeIter = new MergeIterator(iters);
        MetaFile newFile = new MetaFile(path + File.separator + "merge", false);
        int count = 0;
        while (mergeIter.hasNext()) {
            Entry<Long, byte[]> e = mergeIter.next();
            newFile.write(e.getKey(), e.getValue());
            count++;
            //if(count%1000==0)
            //	System.out.println("write key "+e.getKey());
        }
        newFile.close();
        System.out.println("count=" + count);
        synchronized (fileList) {
            for (MetaFile f : metaFiles) {
                f.close();
            }
            File file = new File(path + File.separator + "merge");
            file.renameTo(new File(lastFile));
            file = new File(path + File.separator + "merge.index");
            file.renameTo(new File(lastFile + ".index"));
            for (MetaFile tmp : files) {
                if (!tmp.getPath().equals(lastFile)) {
                    fileList.remove(tmp);
                    file = new File(tmp.getPath());
                    file.delete();
                    file = new File(tmp.getPath() + ".index");
                    file.delete();
                }
            }
            fileList.removeFirst();
            fileList.addFirst(new MetaFile(lastFile, true));
            fileList.notifyAll();
        }
        System.out.println("merge complete");
        Configuration.setBusy(false);
    }

    public class MergeIterator implements Iterator<Entry<Long, byte[]>> {

        private Entry<Long, byte[]>[] heads;
        private Iterator<Entry<Long, byte[]>>[] iterators;

        @SuppressWarnings("unchecked")
        public MergeIterator(Iterator<Entry<Long, byte[]>>[] iterators) {
            this.iterators = iterators;
            heads = new Entry[iterators.length];
            for (int i = 0; i < heads.length; i++) {
                readHead(i);
            }
        }

        public boolean hasNext() {
            for (int i = 0; i < heads.length; i++)
                if (heads[i] != null)
                    return true;
            return false;
        }

        public Entry<Long, byte[]> next() {
            Entry<Long, byte[]> ret = null;
            int index = -1;
            for (int i = 0; i < heads.length; i++) {
                if (heads[i] != null) {
                    if (ret == null) {
                        ret = heads[i];
                        index = i;
                    } else if (heads[i].getKey() < ret.getKey()) {
                        ret = heads[i];
                        index = i;
                    } else if (heads[i].getKey().equals(ret.getKey())) {
                        // discard an old and duplicate record
                        readHead(index);
                        ret = heads[i];
                        index = i;
                    }
                }
            }
            if (index != -1)
                readHead(index);
            return ret;
        }

        private void readHead(int i) {
            if (iterators[i].hasNext())
                heads[i] = iterators[i].next();
            else
                heads[i] = null;
        }

        public void remove() {
            throw new RuntimeException("Not implemented");
        }
    }

    public class MergeThread extends Thread {
        private boolean running = true;

        public void terminate() {
            running = false;
            this.interrupt();
        }

        @Override
        public void run() {
            while (running) {
                synchronized (fileList) {
                    while (fileList.size() < 20) {//Configuration.totalSize/Configuration.blockSize/Configuration.metaBufferSize){
                        try {
                            fileList.wait();
                        }
                        catch (InterruptedException e) {
                            //e.printStackTrace();
                            System.out.println("MergeThread Exit");
                            return;
                        }
                    }
                }
                mergeFile();
            }
            System.out.println("MergeThread Exit");
        }

    }

    public static void main(String[] args) {
        MetaStorage storage = new MetaStorage("/mnt", "meta");
        Random r = new Random();
        long startTime = System.currentTimeMillis();

        for (long i = 0; i < 25000000L; i++) {
            byte[] data = storage.read(i);
            if (data == null) {
                System.out.println("cannot read " + i);
                break;
            }
            ByteBuffer b = ByteBuffer.wrap(data);
            long value = b.getLong();
            if (value != i) {
                System.out.println("wrong value " + i + " " + value);
                break;
            }
            if (i % 1000 == 0)
                System.out.println(i + " " + value);
        }

        for (long i = 0; i < 250000000L; i++) {
            long key = Math.abs(r.nextLong()) % 25000000L;
            byte[] data = new byte[8];
            ByteBuffer b = ByteBuffer.wrap(data);
            b.putLong(key);
            storage.write(key, data);
            /*if(i%25000000==24999999){
                   storage.flushFile();
                   storage.mergeFile();
               }*/
        }

        for (long i = 0; i < 25000000L; i++) {
            long key = Math.abs(r.nextLong()) % 25000000L;
            byte[] data = storage.read(key);
            if (data == null) {
                System.out.println("cannot read " + key);
                break;
            }
            ByteBuffer b = ByteBuffer.wrap(data);
            long value = b.getLong();
            if (value != key) {
                System.out.println("wrong value " + key + " " + value);
                break;
            }
            if (i % 1000 == 0)
                System.out.println(key + " " + value);
        }
        //storage.close();
        System.out.println("time = " + (System.currentTimeMillis() - startTime));
        //storage.flushFile();
        //storage.mergeFile();
    }

}
